package com.tuanzhang.zhw.rpc.register;

import com.tuanzhang.zhw.rpc.client.RpcClient;
import com.tuanzhang.zhw.rpc.config.consumer.CachedService;
import com.tuanzhang.zhw.rpc.config.consumer.ConsumerConfig;
import com.tuanzhang.zhw.rpc.register.zookeeper.AbstractRpcRegistry;
import com.tuanzhang.zhw.rpc.register.zookeeper.ChangedEvent;
import com.tuanzhang.zhw.rpc.register.zookeeper.ZookeeperClient;
import com.tuanzhang.zhw.rpc.utils.FastJsonConvertUtil;
import org.apache.commons.lang3.StringUtils;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 客户端注册&服务 & 服务节点监听
 * 服务发现核心类 ，监听zk的数据节点发生变更 通知
 */
public class RpcRegistryConsumerService extends AbstractRpcRegistry implements NodeListener {

    private ZookeeperClient zookeeperClient;
    private final ReentrantLock LOCK = new ReentrantLock();

    /**
     * key  com.tuanzhang.zhw.rpc.service.UserService:v1.0
     * value  ip:port
     */
    private ConcurrentHashMap<String, List<CachedService>> CACHED_SERVICES = new ConcurrentHashMap<>();

    /**
     * key com.tuanzhang.zhw.rpc.service.UserService:v1.0
     * value ConsumerConfig
     */
    private ConcurrentHashMap<String, ConsumerConfig<?>> CACHED_CONSUMER_CONFIGS = new ConcurrentHashMap<>();

    public RpcRegistryConsumerService(ZookeeperClient client) throws Exception {
        this.zookeeperClient = client;

        //初始化根节点
        if (!zookeeperClient.checkExists(ROOT_PATH)) {
            zookeeperClient.addPersistentNode(ROOT_PATH, ROOT_VALUE);
        }

        // 传入根节点 ROOT_PATH 去监听下一个以及子节点
        /**
         * 	/zhw-rpc   --->  	zhe-rpc-1.0.0
         * 		/com.tuanzhang.zhw.rpc.service.UserService     => 监听根结点只能得到这里的信息变更
         * 			/providers
         * 				/192.168.11.101:5678
         * 				/192.168.11.102:5679
         * 			/consumers
         * 				/192.168.11.103
         *
         */
        this.zookeeperClient.listener4ChildrenPath(ROOT_PATH, this);

    }


    /**
     *  根据服务获取 客户端
     * @param interfaceName
     * @param interfaceVersion
     * @return
     */
    public ConsumerConfig<?> getConsumer(String interfaceName, String interfaceVersion) {
       return CACHED_CONSUMER_CONFIGS.get(interfaceName + ":" + interfaceVersion);
    }


    @Override
    public void nodeChanged(ZookeeperClient client, ChangedEvent event) throws Exception {
        //节点
        String path = event.getPath();
        // 数据信息
        String data = event.getData();
        // 监听类型
        ChangedEvent.Type type = event.getType();

        // 添加
        if (ChangedEvent.Type.CHILD_ADDED == type) {
            String[] pathArray = null;
            if (!StringUtils.isBlank(path) && (pathArray = path.substring(1).split("/")).length == 2) {

                // 对节点下的直接子节点继续监听
                // zhw-rpc/com.tuanzhang.zhw.rpc.service.UserService:v1.0
                this.zookeeperClient.listener4ChildrenPath(path, this);
            }
            // 监听 zhw-rpc/com.tuanzhang.zhw.rpc.service.UserService:v1.0/providers
            if (!StringUtils.isBlank(path) && (pathArray = path.substring(1).split("/")).length == 3) {
                this.zookeeperClient.listener4ChildrenPath(path, this);
            }
            // 监听 zhw-rpc/com.tuanzhang.zhw.rpc.service.UserService:v1.0/providers/127.0.0:8082
            if (!StringUtils.isBlank(path) && (pathArray = path.substring(1).split("/")).length == 4) {

                LOCK.lock();
                try {
                    /**
                     * [0] zhw-rpc
                     * [1] com.tuanzhang.zhw.rpc.service.UserService:v1.0
                     * [2] providers
                     * [3] 127.0.0:8082
                     *
                     *
                     */
                    String interfaceNameService = pathArray[1];
                    String[] arrays = interfaceNameService.split(":");
                    String interfaceName = arrays[0];
                    String address = pathArray[3];
                    Map<String, String> instanceMap = FastJsonConvertUtil.convertJSONToObject(data, Map.class);

                    int weight = Integer.valueOf(instanceMap.get("weight"));
                    CachedService cachedService = new CachedService(address, weight);

                    List<CachedService> cachedServiceList = CACHED_SERVICES.get(interfaceNameService);
                    if (cachedServiceList == null) {
                        CopyOnWriteArrayList<CachedService> servicesList = new CopyOnWriteArrayList<>();
                        servicesList.add(cachedService);
                        CACHED_SERVICES.put(interfaceNameService, servicesList);

                        // 初始化 consumerConfig,建立连接 生成代理对象
                        ConsumerConfig<?> consumerConfig = new ConsumerConfig<>();
                        consumerConfig.setInterfaceName(interfaceName);

                        CopyOnWriteArrayList<String> urls = new CopyOnWriteArrayList<>();

                        //对服务提供者address 进行加权
                        for (int i = 0; i < weight; i++) {
                            urls.add(address);
                        }

                        consumerConfig.setUrl(urls);
                        consumerConfig.initRpcClient();

                        CACHED_CONSUMER_CONFIGS.put(interfaceNameService, consumerConfig);
                    }

                    // 更新缓存
                    else {

                        cachedServiceList.add(cachedService);
                        ConsumerConfig<?> consumerConfig = CACHED_CONSUMER_CONFIGS.get(interfaceNameService);
                        RpcClient rpcClient = consumerConfig.getClient();
                        CopyOnWriteArrayList<String> urls = new CopyOnWriteArrayList<>();
                        for (CachedService service : cachedServiceList) {
                            // 根据权重对address进行加权 => 这里粗略地设置为权重个address
                            for (int i = 0; i < service.getWeight(); i++) {
                                urls.add(service.getAddress());
                            }
                        }
                        //更新 consumer 服务列表
                        consumerConfig.setUrl(urls);

                        //更新连接
                        rpcClient.updateConnectedServer(urls);

                    }

                } finally {
                    LOCK.unlock();
                }

            }


        }
    }

}
